package com.zhang.third.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import javax.swing.plaf.PanelUI;

/**
 * @title: 内连接查询
 * @author: zhang
 * @date: 2022/4/9 16:36
 */
public class Example9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Long>> left = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                });

        SingleOutputStreamOperator<Tuple2<String, Long>> right = env
                .socketTextStream("localhost", 9998)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                });

        left
                .keyBy(r -> r.f0)
                .connect(right.keyBy(r -> r.f0))
                .process(new CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    private ListState<Tuple2<String, Long>> leftHistory;
                    private ListState<Tuple2<String, Long>> rightHistory;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        leftHistory = getRuntimeContext().getListState(
                                new ListStateDescriptor<Tuple2<String, Long>>(
                                        "left", Types.TUPLE(Types.STRING, Types.LONG)
                                )
                        );

                        rightHistory = getRuntimeContext().getListState(
                                new ListStateDescriptor<Tuple2<String, Long>>(
                                        "right", Types.TUPLE(Types.STRING, Types.LONG)
                                )
                        );
                    }

                    @Override
                    public void processElement1(Tuple2<String, Long> value, CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        leftHistory.add(value);
                        for (Tuple2<String, Long> right : rightHistory.get()) {
                            out.collect(value + "=>" + right);
                        }
                    }

                    @Override
                    public void processElement2(Tuple2<String, Long> value, CoProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        rightHistory.add(value);
                        for (Tuple2<String, Long> left : leftHistory.get()) {
                            out.collect(left + "=>" + value);
                        }
                    }
                })
                .print();

        env.execute();
    }
}
